package com.aotain.jupiter.sqlinjection;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.log4j.Logger;

import com.aotain.common.mongo.DataColumn;
import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.jupiter.flink.statis.AbnStatisTuple;

public class SQLInjectionDetect implements FlatMapFunction<Tuple2<String,SQLInjectionTuple>, AbnStatisTuple>{

	/**
	 * 一分钟内发现相同SQL注入行为的次数
	 */
	private int threshold = 3;
	
	/**
	 * 
	 */
	private static final long serialVersionUID = 6727956188693858262L;

	@Override
	public void flatMap(Tuple2<String, SQLInjectionTuple> arg0,
			Collector<AbnStatisTuple> arg1) throws Exception {
		// TODO Auto-generated method stub
		SQLInjectionTuple tuple = arg0.f1;
		int urlPVNum = tuple.getCounter();
		
		if(urlPVNum >= threshold){
			System.out.println("SQLInjection target >>>>>>>>> "+arg0.f0 + " >>>>>>>>> matchNum:" 
					+  urlPVNum);
			
			String dip = tuple.getDestIP();
			String dport = tuple.getDestPort();
			String sip = tuple.getSourceIP();
			String sport = tuple.getSourcePort();
			String gis = new String(Base64.decodeBase64(tuple.getGIS()));
			String type = tuple.getType();
			String idc = tuple.getIDC();
			String[] gisArray = gis.split("#");
			String destAreaName = "";
			String destGis = "";
			String sourceAreaName = "";
			String sourceGis = "";
			String sourceAreaCountry = "";
			String sourceAreaId = "";
			String sourceProvinceName = "";
			if(gisArray.length > 1) {
				destAreaName = gisArray[0];
				destGis = gisArray[1];
				sourceAreaName = gisArray[2];
				sourceGis = gisArray[3];
				sourceAreaCountry = gisArray[4];
				sourceAreaId = gisArray[5];
				sourceProvinceName =  gisArray[6].trim(); //如果省为空，精确到国家
			} else {
				Logger.getLogger(SQLInjectionDetect.class).warn("GIS Object is null: " + gis);
			}
			
//				String sourceAreaCityId = gisArray[7];
//				String sourceAreaProvinceId = gisArray[8];
		
			String sAbnormalTable = "METIS_ABNORMAL_LOG";
				
			String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(System.currentTimeMillis()+15000));
				//源IP发起的攻击行为
			String rowKeyAbnormal = sip + "_"  + accesstime + "_" + dip + "_" + dport + "_8";
			
			MongoImportTool importtool = MongoImportTool.getInstance();
			List<DataColumn> row = new ArrayList<DataColumn>();
			row.add(new DataColumn("ROWKEY", rowKeyAbnormal)); //主键字段，用于更新记录
			row.add(new DataColumn("SOURCEIP", sip));
			row.add(new DataColumn("DESTPORT", dport));
			row.add(new DataColumn("ACCESSTIME", accesstime));
			row.add(new DataColumn("ABRNORMAL", "8"));
			row.add(new DataColumn("DESTIP", dip));
			row.add(new DataColumn("SOURCEAREA", sourceAreaName));
			row.add(new DataColumn("SOURCEGEO", sourceGis));
			row.add(new DataColumn("SOURCECOUNTRY", sourceAreaCountry));
			row.add(new DataColumn("DESTAREA", destAreaName));
			row.add(new DataColumn("DESTGEO", destGis));
			row.add(new DataColumn("ATTNUM", urlPVNum));
			row.add(new DataColumn("DESC", "SQL Injection:" + tuple.getURL()));
			row.add(new DataColumn("EVALUATE", "40"));
			row.add(new DataColumn("ENDTIME", accesstime));
			if(sourceProvinceName.equals("")) 
				sourceProvinceName = sourceAreaCountry;
			row.add(new DataColumn("PROVINCE", sourceProvinceName));
			row.add(new DataColumn("UPSTREAMOCTETS", tuple.getUpStreamOctets()));
			row.add(new DataColumn("UPSTREAMPACKET", tuple.getUpStreamPacket()));
				
			importtool.InsertRowData(sAbnormalTable, row);
			
			AbnStatisTuple abnTuple = new AbnStatisTuple();
			abnTuple.setSourceIP(sip);
			abnTuple.setType("SQLINJECTION");
			abnTuple.setUpStreamOctets(tuple.getUpStreamOctets());
			arg1.collect(abnTuple);
//			MongoSinkObject sinkObj = new MongoSinkObject();
//			sinkObj.setTableName(sAbnormalTable);
//			sinkObj.setDmlType(DML.INSERT);
//			sinkObj.setImportData(row);
//			arg1.collect(sinkObj);
				
//				Date current = new Date();
//				SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
//				SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
//				SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
//		    	String dateStr_h = sdf_h.format(current);
//				String dateStr_d = sdf_d.format(current);
//				String dateStr_hour = sdf_hour.format(current);
//				
//				collector.emit(new Values(dip,sip,dport,sport,dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,
//		    			sourceProvinceName, urlPVNum, dateStr_hour,"SQLINJECTION" , idc, sourceAreaCountry,(long)nUpStreamOctets,(long)nUpstreampacket));
//		    	
				
			
		}
		
		
	}

}
